/*
 * Readaheadmanager.cpp
 *
 *  Created on: 01.02.2013
 *      Author: mario
 */

#include "Readahead_manager.h"

// omnibus
#include "Filetransfer.h"
#include "UI_multiplexer.h" // XXX debug

// ariba
#include <ariba/utility/system/SystemQueue.h>


////////////////////////////////////////////////////////////////////////

/* 
 * classes:
 *   - read_ahead_thread
 *   - Read_File
 *   - Readahead_manager
 */

////////////////////////////////////////////////////////////////////////


//** Thread class **//
class read_ahead_thread
{
public:
    typedef Readahead_manager::FileList FileList;
    
    
    // constructor
    read_ahead_thread(shared_ptr<Readahead_manager> manager) :
        manager(manager)
    {
    }
    
    // * thread function *
    void operator()()
    {
        // * thread loop *
        while ( ! manager->stop_thread )
        {
            sleep_on_idle();
            cleanup_canceled_files();
            get_active_files();
    
            // as long as we have active files
            while ( active_files.size() > 0 && !manager->stop_thread )
            {
                // read one chunk from each active file
                for (FileList::iterator it = active_files.begin(); it != active_files.end() && !manager->stop_thread; ++it)
                {
                    // * read chunk *
                    bool still_active = read_chunk(*it);
                    
                    // trigger callback if enabled
                    if ( (*it)->callback_enabled )
                    {
                        (*it)->deactivate_callback();
                        manager->trigger_callback((*it)->fileptr);
                    }

                    // remove file if it's inactive now
                    if ( ! still_active )
                    {
                        // remove from active files
                        it = active_files.erase(it);
                    }
                }
                
                // refresh active file list, if file list was updated
                if ( manager->new_work )  // read only access --> no locking necessary
                {
                    cleanup_canceled_files();
                    get_active_files();
                }
            }
        } // [ while ( ! manager->stop_thread ) ]

        
        // clean up
        cleanup_canceled_files();
        
        // last thing to do: mark as stopped
        manager->thread_is_running = false;
        delete manager->thread;                 // TODO is this okay here..?
    }
    
    
private:
    /* sleep while there's no work */
    void sleep_on_idle()
    {
        boost::unique_lock<boost::mutex> lock(manager->filelist_mutex);
        
        while( ! manager->new_work )
        {
            manager->thread_is_sleeping = true;
            manager->work_condvar.wait(lock);
            manager->thread_is_sleeping = false;
        }
    }
    
    
    void cleanup_canceled_files()
    {
        FileList tmp;
        
        // move all elements out of the "locked" cancel_list
        // LOCKED SCOPE
        {
            boost::lock_guard<boost::mutex> lock(manager->cancel_list_mutex);
            
            tmp.splice(tmp.begin(), manager->cancel_list);
        }
        
        // delete all canceled elements (in uncritical section)
        for (FileList::iterator it = tmp.begin(); it != tmp.end(); ++it)
        {
            delete (*it);
            
            // NOTE: we do not "erase" here since this tmp list is going to die anyway.
        }
    }
    
    
    void get_active_files()
    {
        // LOCKED SCOPE
        {
            boost::lock_guard<boost::mutex> lock(manager->filelist_mutex);

            active_files = manager->files;
            
            manager->new_work = false;
        }
        
        
        // remove inactive files
        for (FileList::iterator it = active_files.begin(); it != active_files.end(); ++it)
        {
            // check state
            bool err = (*it)->get_state() == Read_File::ERROR; 
            
            // remove all files with already filled buffers
            if ( (*it)->get_buffer_size() >= FILE_QUEUE_SIZE || err )
            {
                it = active_files.erase(it);
            }
        }
    }

    
    // reads data from file into given buffer
    bool read_chunk(Read_File* from_file)
    {
        Read_File::State state = from_file->get_state();
        
        if ( state == Read_File::CANCEL )
        {
            return false;  // false: file not active any more!
        }
        
        
        // open file if necessary
//        file.exceptions(ofstream::failbit);  // XXX ??
        if ( ! from_file->file.is_open() )
        {
            if ( state == Read_File::NEW )
            {
                // open
                from_file->file.open(from_file->filename.c_str(), ios::binary);
                
                // seek to start position
                from_file->file.seekg(from_file->start_position);
                
                from_file->set_state(Read_File::READING);
            }
            else
            {
                from_file->set_state(Read_File::ERROR);
                
                return false;  // false: file not active any more!
            }
        }
        
        
        // * read from file *
        if ( ! from_file->file.eof() && ! from_file->file.fail() )
        {
            // new buffer
            file_chunk* chunk = new file_chunk(CHUNKSIZE);

            // store position
            chunk->position = from_file->file.tellg();

            // * read chunk *
            from_file->file.read( (char*) chunk->data.mutable_data(), CHUNKSIZE );

            // how much has been read?
            chunk->size = from_file->file.gcount();
            
            // BRANCH: data was actually read  --> push into buffer
            if ( chunk->size > 0 )
            {
                // resize buffer if chunk.size < CHUNKSIZE
                if ( chunk->size < CHUNKSIZE )
                {
                    chunk->data.resize(chunk->size);   // TODO this could be quite inefficient! 
                }

                // * put chunk into buffer * (NOTE: it's accessible from other threads now!)
                size_t queue_size = from_file->append_chunk(chunk);  // ownership of "chunk" transferred to buffer 
                
                // return true if file is still active
                return queue_size < FILE_QUEUE_SIZE; 
            }
            // BRANCH: no data read --> delete chunk
            else
            {
                delete chunk;
                chunk = 0;
            }
        }
        // TODO else: error handling?  (finished or error..?)


        // didn't return true yet, no more data to read
        from_file->file.close();

        // read finished
        from_file->set_state(Read_File::FINISHED);
        _move_to_finished1(from_file);
        
        return false;
    }
    
    
    void _move_to_finished1(Read_File* file)
    {
        // LOCKED SCOPE  (2x)
        boost::lock_guard<boost::mutex> lock(manager->filelist_mutex);
        boost::lock_guard<boost::mutex> lock2(manager->finished1_mutex);
        
        FileList& file_list = manager->files;
        
        // find (iterator of) read_file object in "files" and move it to "finished_files1"
        for (FileList::iterator it = file_list.begin(); it != file_list.end(); ++it)
        {
            if ( (*it) == file )
            {
                manager->finished_files1.splice(manager->finished_files1.begin(), file_list, it);
                
                break;
            }
        }
    }

    
private:
    shared_ptr<Readahead_manager> manager;
    FileList active_files;
};

//** [Thread class] **//


////////////////////////////////////////////////////////////////////////


//** Read_File class **//

/*
 *  constructor
 *    ---> all variables accessible, since this object is not known by the read_ahead thread, yet!
 */
Read_File::Read_File(FilePtr f, uint64_t start_pos)  :
        fileptr(f),
        filename(f->get_path()), 
        start_position(start_pos),
        state(NEW),
        callback_enabled(false)
{
}


Read_File::~Read_File()
{
    // close file
    if ( file.is_open() )
        file.close();
    
    // delete all chunks in buffer
    for (deque<file_chunk*>::iterator it = data_buffer.begin(); it != data_buffer.end(); ++it)
        delete (*it);
    
    
    // XXX debug
//    UI.write_debug("/// MARIO: ~Read_File(): " + filename);
}


size_t Read_File::get_buffer_size()
{
    // LOCKED SCOPE  [buffer_lock]
    {
        boost::lock_guard<boost::mutex> lock(buffer_lock);

        return data_buffer.size();
    }
}

size_t Read_File::append_chunk(file_chunk* data)
{
    // LOCKED SCOPE  [buffer_lock]
    {
        boost::lock_guard<boost::mutex> lock(buffer_lock);

        data_buffer.push_back(data);
        
        return data_buffer.size();
    }
}

file_chunk* Read_File::get_next_chunk()
{
    // LOCKED SCOPE  [buffer_lock]
    {
        boost::lock_guard<boost::mutex> lock(buffer_lock);

        // BRANCH: buffer empty --> return NULL
        if ( data_buffer.empty() )
            return NULL;
        
        // BRANCH: buffer not empty --> return first element
        file_chunk* chunk = data_buffer.front();
        data_buffer.pop_front();
        
        return chunk;  // ownership of "chunk" transferred to caller
    }
}

void Read_File::set_state(State state)
{
    // LOCKED SCOPE  [state_lock]
    {
        boost::lock_guard<boost::mutex> lock(state_lock);

        this->state = state;
    }
}

Read_File::State Read_File::get_state()
{
    // LOCKED SCOPE  [state_lock]
    {
        boost::lock_guard<boost::mutex> lock(state_lock);

        return this->state;
    }
}

void Read_File::activate_callback()
{
    // LOCKED SCOPE  [callback_lock]
    {
        boost::lock_guard<boost::mutex> lock(callback_lock);

        this->callback_enabled = true;
    }
}

void Read_File::deactivate_callback()
{
    // LOCKED SCOPE  [callback_lock]
    {
        boost::lock_guard<boost::mutex> lock(callback_lock);

        this->callback_enabled = false;
    }
}

//** [Read_File class] **//


////////////////////////////////////////////////////////////////////////



Readahead_manager::Readahead_manager() throw()  :
    new_work(false),
    thread_is_sleeping(false),
    stop_thread(false),
    thread_is_running(false)
{
}

Readahead_manager::~Readahead_manager() throw()
{
    assert( ! thread_is_running );

    // all queues should be empty
    //    ---> clean them anyway, just in case..
    for (FileList::iterator it = finished_files2.begin(); it != finished_files2.end(); ++it)
        delete (*it);
    for (FileList::iterator it = finished_files1.begin(); it != finished_files1.end(); ++it)
        delete (*it);
    for (FileList::iterator it = files.begin(); it != files.end(); ++it)
        delete (*it);
    
    
    // TODO: check if this is ever called
    //   ---> and therefore if the thread object is ever destroyed
    UI.write_debug("/// MARIO: ~Readahead_manager()");
    cout << endl << "/// MARIO: ~Readahead_manager()" << endl;
    cerr << endl << "/// MARIO: ~Readahead_manager()" << endl;
}


void Readahead_manager::start(Filetransfer* filetransfer) throw()
{
    assert( ! thread_is_running );
    
    // register filetransfer object for callbacks
    this->filetransfer = filetransfer;
    
    // * start thread *
    thread_is_running = true;
    read_ahead_thread t(shared_from_this());
    thread = new boost::thread(t);
}

void Readahead_manager::stop() throw()
{
    assert( thread_is_running );
    
    // cancel all files
    //   threat all access invalid is now
    //   ---> move all read_files to cancel_list
    //   ---> let the read_ahead_thread do the cleanup after quitting it's main loop
    //
    // LOCKED SCOPE (3x)
    {
        boost::lock_guard<boost::mutex> lock(filelist_mutex);
        boost::lock_guard<boost::mutex> lock2(cancel_list_mutex);
        boost::lock_guard<boost::mutex> lock3(finished1_mutex);

        cancel_list.splice(cancel_list.begin(), finished_files2);
        cancel_list.splice(cancel_list.begin(), finished_files1);
        cancel_list.splice(cancel_list.begin(), files);
        
        new_work = true;
    }
    
    // signal the thread to stop
    stop_thread = true;
    
    // maybe we have to wake the thread
    if ( thread_is_sleeping )
        work_condvar.notify_one();
}


void Readahead_manager::add_file(FilePtr file, uint64_t position) throw()
{
    Read_File* read_file = new Read_File(file, position);
    
    // LOCKED SCOPE  [this->files]
    {
        boost::lock_guard<boost::mutex> lock(filelist_mutex);

        // insert into file-list
        //    ---> now it's accessible from the read_ahead_thread (after this lock is released)
        files.push_front(read_file);
        
        new_work = true;
    }
    
    // wake the read_ahead thread, if it's not already running
    if ( thread_is_sleeping )
        work_condvar.notify_one();
}


void Readahead_manager::remove_file(FilePtr file) throw(not_found_exception)
{
    Read_File* read_file = NULL;
    
    // first:
    //   - remove file from finished_files2 (if it's in there)
    read_file = _remove_file_from(finished_files2, file);
    
    
    // second:
    //   if not found, yet:
    //   - remove file from finished_files1 (if it's in there)
    //   - move all (remaining) elements from finished1 to finished2
    //
    // LOCKED SCOPE  [finished1_mutex]
    if ( read_file == NULL )
    {
        boost::lock_guard<boost::mutex> lock(finished1_mutex);
        
        // remove file from finished_files1 (if it's in there)
        read_file = _remove_file_from(finished_files1, file);
        
        // move all elements from finished_files1 to finished_files2
        finished_files2.splice(finished_files2.begin(), finished_files1);
    }
    
    
    // if found (in first or second):
    //   ---> delete read_file object (which clears the buffers).
    //
    //   return;
    if ( read_file )
    {
        delete read_file;
        
        return;
    }

    
    // third:
    //  if still not found, yet:
    //   - find file in (non-finished) files
    //   - set status to CANCEL
    //   - move it to canel_list
    //       ---> don't delete it!
    //
    //   - wake read_ahead_thread if it's sleeping
    // 
    // LOCKED SCOPE  [filelist_mutex]
    {
        boost::lock_guard<boost::mutex> lock(filelist_mutex);
        
        for (FileList::iterator it = files.begin(); it != files.end(); ++it)
        {
            if ( (*it)->fileptr == file )
            {
                read_file = (*it);
                read_file->set_state(Read_File::CANCEL);
                
                // LOCKED SCOPE  [cancel_list_mutex]
                {
                    boost::lock_guard<boost::mutex> lock(cancel_list_mutex);
                    
                    // read_file may still be in use by read_ahed_thread ---> move to cancel queue
                    cancel_list.splice(cancel_list.begin(), files, it);
                }

                // wake-up condition
                new_work = true;
                
                break;
            }
        }
    }
    
    // wake the read_ahead thread, if it's not already running
    if ( new_work && thread_is_sleeping )
    {
        work_condvar.notify_one();
    }
    
    
    // exception if file still not found
    if ( read_file == NULL )
        throw not_found_exception("Readahead_manager (in remove_file): File not found!");
}


file_chunk Readahead_manager::get_next_chunk(FilePtr file) throw(
        End_of_File_Notification,
        Wait_for_it_Notification,
        not_found_exception,
        Bad_File_Exception)
{
    Read_File* read_file = NULL;
    file_chunk* chunk = NULL;
    
    
    // first: search in finished2 files
    try
    {
        _get_next_chunk_from(finished_files2, file, read_file, chunk);
    }
    catch ( const Buffer_Empty_Notification& e )
    {
        //throw end_of_file_exception;
        throw End_of_File_Notification();
    }
    
    
    // second:
    //   - search in finished1
    //   - move all elements from finished1 to finished2
    //
    // LOCKED SCOPE  [finished1_mutex]
    if ( read_file == NULL )
    {
        boost::lock_guard<boost::mutex> lock(finished1_mutex);
        
        // search in finished1 (NOTE: access to finished1 must be locked with finished1_mutex])
        try
        {
            _get_next_chunk_from(finished_files1, file, read_file, chunk);
        }
        catch ( Buffer_Empty_Notification& e )
        {
//            throw end_of_file_exception;
            throw End_of_File_Notification();
        }

        // move all elements from finished_files1 to finished_files2
        finished_files2.splice(finished_files2.begin(), finished_files1);
    }
        
    
    // third:
    //   - search in (non-finished) filelist
    //
    // LOCKED SCOPE  [filelist_mutex]
    if ( read_file == NULL )
    {
        boost::lock_guard<boost::mutex> lock(filelist_mutex);
        
        try
        {
            _get_next_chunk_from(files, file, read_file, chunk);
        }
        catch ( Buffer_Empty_Notification& e )
        {
            read_file->activate_callback();
            
//                throw wait_for_it_exception;
            throw Wait_for_it_Notification();
        }
        
        if ( chunk != NULL )
            new_work = true;
    }

    
    // wake the read_ahead thread, if it's not already running
    if ( new_work && thread_is_sleeping )
    {
        work_condvar.notify_one();
    }

    
    // ERR: throw if file not found -- should never happen -- doesn't have to be caught.
    if ( read_file == NULL )
        throw not_found_exception("Readahead_manager: File not found!");
    
    
    // returns chunk-object by value and deletes our own copy
    std::auto_ptr<file_chunk> auto_chunk(chunk);
    chunk = NULL;
            
    return *auto_chunk;
}


inline void Readahead_manager::_get_next_chunk_from(FileList& file_list, FilePtr file, Read_File*& f, file_chunk*& chunk)
        throw(Bad_File_Exception, Buffer_Empty_Notification)
        
{
    for (FileList::iterator it = file_list.begin(); it != file_list.end(); ++it)
    {
        if ( (*it)->fileptr == file )
        {
            f = *it;

            // ERR: bad file, an error occurred reading this file
            if ( f->get_state() == Read_File::ERROR )
//                throw bad_file;
                throw Bad_File_Exception();

            
            // * get chunk *
            chunk = f->get_next_chunk();   // ownership of "chunk" transferred to caller
            
            
            // check if buffer is empty
            if ( chunk == NULL )
//                throw buffer_empty_exception;
                throw Buffer_Empty_Notification();
            
            break;
        }
    }
}

inline Read_File* Readahead_manager::_remove_file_from(FileList& file_list, FilePtr file)
{
    Read_File* ret = NULL;
    
    for (FileList::iterator it = file_list.begin(); it != file_list.end(); ++it)
    {
        if ( (*it)->fileptr == file )
        {
            ret = (*it);
            file_list.erase(it);
            
            return ret;
        }
    }

    return NULL;
}

void Readahead_manager::trigger_callback(FilePtr file) const
{
    // don't trigger callbacks when this thread should be stopped
    //   ---> SystemQueue may be stopped or "Filetransfer* filetransfer" could be invalid
    if ( stop_thread )
        return;
    
    
    // * insert callback into System Queue *
    SystemQueue::instance().scheduleCall(
            boost::bind(
                    &Filetransfer::data_ready_callback,
                    filetransfer,
                    file)
        );
}
